/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.client.impl.consumer;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.client.consumer.PullCallback;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.hook.FilterMessageContext;
import org.apache.rocketmq.client.hook.FilterMessageHook;
import org.apache.rocketmq.client.impl.CommunicationMode;
import org.apache.rocketmq.client.impl.FindBrokerResult;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.filter.ExpressionType;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.sysflag.PullSysFlag;
import org.apache.rocketmq.remoting.exception.RemotingException;


public class PullAPIWrapper {
    private final InternalLogger log = ClientLogger.getLog();
    private final MQClientInstance mQClientFactory;
    private final String consumerGroup;
    private final boolean unitMode;
    private ConcurrentMap<MessageQueue, AtomicLong/* brokerId */> pullFromWhichNodeTable =
            new ConcurrentHashMap<MessageQueue, AtomicLong>(32);
    private volatile boolean connectBrokerByUser = false;
    private volatile long defaultBrokerId = MixAll.MASTER_ID;
    private Random random = new Random(System.currentTimeMillis());
    private ArrayList<FilterMessageHook> filterMessageHookList = new ArrayList<FilterMessageHook>();

    public PullAPIWrapper(MQClientInstance mQClientFactory, String consumerGroup, boolean unitMode) {
        this.mQClientFactory = mQClientFactory;
        this.consumerGroup = consumerGroup;
        this.unitMode = unitMode;
    }

    /**
     * 处理pullResult，进行消息解码、过滤以及设置其他属性的操作
     * @param mq MSGQUEUE
     * @param pullResult 拉取消息的结果
     * @param subscriptionData 订阅的数据
     * @return
     */
    public PullResult processPullResult(final MessageQueue mq, final PullResult pullResult,
                                        final SubscriptionData subscriptionData) {
        // 返回结果
        PullResultExt pullResultExt = (PullResultExt) pullResult;
        /*
         * 1 更新下次拉取建议的brokerId，下次拉取消息时从pullFromWhichNodeTable中直接取出
         */
        this.updatePullFromWhichNode(mq, pullResultExt.getSuggestWhichBrokerId());
        if (PullStatus.FOUND == pullResult.getPullStatus()) {
            /*
             * 2 对二进制字节数组进行解码转换为java的List<MessageExt>消息集合
             */
            ByteBuffer byteBuffer = ByteBuffer.wrap(pullResultExt.getMessageBinary());
            List<MessageExt> msgList = MessageDecoder.decodes(byteBuffer);

            List<MessageExt> msgListFilterAgain = msgList;
            /*
             * 3 如果存在tag，并且不是classFilterMode，那么按照tag过滤消息，这就是客户端的消息过滤
             *
             * 通过前面的源码可以看到，消息实际上经过了两次过滤，一次是在broekr中，一次是拉取到consumer之后，为什么经过两次过滤呢？
             * 因为broker中的过滤是比较的hashCode值，而hashCode存在哈希碰撞的可能，因此hashCode对比相等之后，还需要在consumer端进行equals的比较，再过滤一次。
             *
             * 为什么服务端不直接进行equals过滤呢？因为tag的长度是不固定的，而通过hash算法可以生成固定长度的hashCode值，这样才能保证
             * 每个consumequeue索引条目的长度一致。而tag的真正值保存在commitLog的消息体中，虽然broker最终会取到commitLog中的消息并返回，
             * 但是获取的一段消息字节数组，并没有进行反序列化为Message对象，因此无法获取真实值，而在consumer端一定会做反序列化操作的，
             * 因此tag的equals比较放在了consumer端。
             */
            if (!subscriptionData.getTagsSet().isEmpty() && !subscriptionData.isClassFilterMode()) {
                msgListFilterAgain = new ArrayList<MessageExt>(msgList.size());
                for (MessageExt msg : msgList) {
                    if (msg.getTags() != null) {
                        // 这采用String#equals方法过滤，而broker端则是比较的tagHash值，即hashCode
                        if (subscriptionData.getTagsSet().contains(msg.getTags())) {
                            msgListFilterAgain.add(msg);
                        }
                    }
                }
            }
            /*
             * 4 如果有消息过滤钩子，那么执行钩子方法，这里可以扩展自定义的消息过滤的逻辑
             */
            if (this.hasHook()) {
                FilterMessageContext filterMessageContext = new FilterMessageContext();
                filterMessageContext.setUnitMode(unitMode);
                filterMessageContext.setMsgList(msgListFilterAgain);
                this.executeHook(filterMessageContext);
            }
            /*
             * 5 遍历过滤通过的消息，设置属性
             */
            for (MessageExt msg : msgListFilterAgain) {
                // 事务消息标识
                String traFlag = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
                // 如果是事务消息，则设置事务id
                if (Boolean.parseBoolean(traFlag)) {
                    msg.setTransactionId(msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
                }
                // 将响应中的最小和最大偏移量存入msg
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MIN_OFFSET,
                        Long.toString(pullResult.getMinOffset()));
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MAX_OFFSET,
                        Long.toString(pullResult.getMaxOffset()));
                // 设置brokerName到msg
                msg.setBrokerName(mq.getBrokerName());
            }
            // 将过滤后的消息存入msgFoundList集合
            pullResultExt.setMsgFoundList(msgListFilterAgain);
        }
        /**
         * 6 因为消息已经被解析了，那么设置消息的字节数组为null，释放内存
         */
        pullResultExt.setMessageBinary(null);
        // 返回
        return pullResult;
    }

    /**
     * 更新下次拉取建议的brokerId
     *
     * @param mq       消息队列
     * @param brokerId 建议的brokerId
     */
    public void updatePullFromWhichNode(final MessageQueue mq, final long brokerId) {
        // 存入pullFromWhichNodeTable集合
        AtomicLong suggest = this.pullFromWhichNodeTable.get(mq);
        if (null == suggest) {
            this.pullFromWhichNodeTable.put(mq, new AtomicLong(brokerId));
        } else {
            suggest.set(brokerId);
        }
    }

    public boolean hasHook() {
        return !this.filterMessageHookList.isEmpty();
    }

    public void executeHook(final FilterMessageContext context) {
        if (!this.filterMessageHookList.isEmpty()) {
            for (FilterMessageHook hook : this.filterMessageHookList) {
                try {
                    hook.filterMessage(context);
                } catch (Throwable e) {
                    log.error("execute hook error. hookName={}", hook.hookName());
                }
            }
        }
    }

    public PullResult pullKernelImpl(
            final MessageQueue mq, // 消息消费队列的元数据信息
            final String subExpression, // 消息订阅子模式，subscribe(topicName, "模式")
            final String expressionType, // "模式"表达式
            final long subVersion, // 版本
            final long offset, // 开始拉取的消息偏移量，pullRequest.getNextOffset()
            final int maxNums, // 拉取的最大消息个数，defaultMQPushConsumer.getPullBatchSize()
            final int sysFlag, // 系统标识，FLAG_COMMIT_OFFSET_SUSPEND
            final long commitOffset, // 内存中当前消息队列commitLog日志中当前的最新偏移量
            final long brokerSuspendMaxTimeMillis, // 允许broker暂停的时间，单位：毫秒，默认15s
            final long timeoutMillis, // 拉取消息的超时时间
            final CommunicationMode communicationMode, // 拉取方式，同步、异步、只发送不关注返回
            final PullCallback pullCallback // 回调逻辑
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // 通过broker名字从内存中找到broker的一些信息，包括broker地址，是否从节点，broker版本
        FindBrokerResult findBrokerResult =
                this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
                        this.recalculatePullFromWhichNode(mq), false);
        if (null == findBrokerResult) {
            // 如果内存中找不到broker信息，那么从NameServer中更新broker的最新信息（有可能是修改了但是还没来得及执行同步）
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
            // 然后再次获取
            findBrokerResult =
                    this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
                            this.recalculatePullFromWhichNode(mq), false);
        }

        if (findBrokerResult != null) {
            // 找到了broker
            {
                // 检查版本,如果小于V4_1_0_SNAPSHOT，那么不支持filter过滤
                if (!ExpressionType.isTagType(expressionType)
                        && findBrokerResult.getBrokerVersion() < MQVersion.Version.V4_1_0_SNAPSHOT.ordinal()) {
                    throw new MQClientException("The broker[" + mq.getBrokerName() + ", "
                            + findBrokerResult.getBrokerVersion() + "] does not upgrade to support for filter message by " + expressionType, null);
                }
            }
            int sysFlagInner = sysFlag;
            // 如果broker是slave节点
            if (findBrokerResult.isSlave()) {
                // 清除拉取消息标记的提交偏移量标志位
                sysFlagInner = PullSysFlag.clearCommitOffsetFlag(sysFlagInner);
            }
            // 封装网络请求体
            PullMessageRequestHeader requestHeader = new PullMessageRequestHeader();
            requestHeader.setConsumerGroup(this.consumerGroup);
            requestHeader.setTopic(mq.getTopic());
            requestHeader.setQueueId(mq.getQueueId());
            requestHeader.setQueueOffset(offset);
            requestHeader.setMaxMsgNums(maxNums);
            requestHeader.setSysFlag(sysFlagInner);
            requestHeader.setCommitOffset(commitOffset);
            requestHeader.setSuspendTimeoutMillis(brokerSuspendMaxTimeMillis);
            requestHeader.setSubscription(subExpression);
            requestHeader.setSubVersion(subVersion);
            requestHeader.setExpressionType(expressionType);

            String brokerAddr = findBrokerResult.getBrokerAddr();
            if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
                // 如果包含filter文件，那么上传到过滤服务器上面
                brokerAddr = computPullFromWhichFilterServer(mq.getTopic(), brokerAddr);
            }
            // 调用MQClient拉取信息，并返回一个拉取结果
            // 注意拉取信息的模式是同步、异步，然后根据不同模式去回调还是直接处理
            PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(
                    brokerAddr,
                    requestHeader,
                    timeoutMillis,
                    communicationMode,
                    pullCallback);

            return pullResult;
        }
        // 如果还是等于 null，那么就表明出问题了，找不到这个broker
        throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
    }

    public long recalculatePullFromWhichNode(final MessageQueue mq) {
        if (this.isConnectBrokerByUser()) {
            return this.defaultBrokerId;
        }

        AtomicLong suggest = this.pullFromWhichNodeTable.get(mq);
        if (suggest != null) {
            return suggest.get();
        }

        return MixAll.MASTER_ID;
    }

    private String computPullFromWhichFilterServer(final String topic, final String brokerAddr)
            throws MQClientException {
        // 获取消息客户端工厂中的主题路由表
        ConcurrentMap<String, TopicRouteData> topicRouteTable = this.mQClientFactory.getTopicRouteTable();
        if (topicRouteTable != null) {
            // 获取路由数据data
            // 一个topic可以有很多个broker
            TopicRouteData topicRouteData = topicRouteTable.get(topic);
            // 获取在这个topic下，设置的一些过滤内容
            List<String> list = topicRouteData.getFilterServerTable().get(brokerAddr);

            if (list != null && !list.isEmpty()) {
                // 随机返回一个
                return list.get(randomNum() % list.size());
            }
        }

        throw new MQClientException("Find Filter Server Failed, Broker Addr: " + brokerAddr + " topic: "
                + topic, null);
    }

    public boolean isConnectBrokerByUser() {
        return connectBrokerByUser;
    }

    public void setConnectBrokerByUser(boolean connectBrokerByUser) {
        this.connectBrokerByUser = connectBrokerByUser;

    }

    public int randomNum() {
        int value = random.nextInt();
        if (value < 0) {
            value = Math.abs(value);
            if (value < 0)
                value = 0;
        }
        return value;
    }

    public void registerFilterMessageHook(ArrayList<FilterMessageHook> filterMessageHookList) {
        this.filterMessageHookList = filterMessageHookList;
    }

    public long getDefaultBrokerId() {
        return defaultBrokerId;
    }

    public void setDefaultBrokerId(long defaultBrokerId) {
        this.defaultBrokerId = defaultBrokerId;
    }
}
